package friends;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


import java.io.IOException;

public class FriendsDriver {
    public static void main(String[] args) throws IOException, InterruptedException {
        //定义输入输出路径
        args=new String[]{"E:\\test\\friends_data\\friends.txt","E:\\test\\temp\\friends_data3","E:\\test\\temp\\friends_data4"};

        //创建配置文件
        Configuration conf = new Configuration();

        //###########################job1######################
        //创建Job1任务
        Job job1 = Job.getInstance(conf);
        //设置Driver反射
        job1.setJarByClass(FirstDriver.class);

        //设置Mapper反射
        job1.setMapperClass(FirstMapper.class);
        //设置Mapper的输出（key,value）类型
        job1.setMapOutputKeyClass(Text.class);
        job1.setMapOutputValueClass(Text.class);

        //设置Reducer反射
        job1.setReducerClass(FirstReducer.class);
        //设置Reducer的输出（key,value）类型
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(Text.class);


        //指定输入输出路径
        FileInputFormat.setInputPaths(job1, new Path(args[0]));
        FileOutputFormat.setOutputPath(job1,new Path(args[1]));

        //###########################job2######################
        //创建Job2任务
        Job job2 = Job.getInstance(conf);

        //设置Driver反射
        job2.setJarByClass(SecondDriver.class);

        //设置Mapper反射
        job2.setMapperClass(SecondMapper.class);
        //设置Mapper的输出（key,value）类型
        job2.setMapOutputKeyClass(Text.class);
        job2.setMapOutputValueClass(Text.class);

        //设置Reducer反射
        job2.setReducerClass(SecondReducer.class);
        //设置Reducer的输出（key,value）类型
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(Text.class);


        //指定输入输出路径
        FileInputFormat.setInputPaths(job2, new Path(args[1]));
        FileOutputFormat.setOutputPath(job2,new Path(args[2]));

        //job控制器
        JobControl jobControl = new JobControl("job");
        ControlledJob controlledJob1 = new ControlledJob(job1.getConfiguration());
        ControlledJob controlledJob2 = new ControlledJob(job2.getConfiguration());
        //依赖
        controlledJob2.addDependingJob(controlledJob1);
        jobControl.addJob(controlledJob1);
        jobControl.addJob(controlledJob2);
        Thread thread = new Thread(jobControl);
        thread.start();
        //提交任务
        while(!jobControl.allFinished()){
            Thread.sleep(1000);

        }
        System.exit(0);
       /* //提交job任务到yarn执行
        boolean isSuccess = job.waitForCompletion(true);
        System.exit(isSuccess?0:-1);*/

    }
}
